package com.itheima.kafka.stream.listener;

import com.itheima.kafka.config.KafkaStreamListener;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.List;

/**
 * 泛型执行执行两种类型
 * Ktable
 * KStream
 */
@Component
public class StreamListener implements KafkaStreamListener<KStream<?,String>> {
    @Override
    public String listenerTopic() {
        return "input_topic";
    }

    @Override
    public String sendTopic() {
        return "out_topic";
    }

    @Override
    public KStream<?,String> getService(KStream<?,String> stream) {
        return stream.groupByKey().windowedBy(TimeWindows.of(50000))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        return "啥也没有";
                    }
                }, new Aggregator<Object, String, String>() {
                    @Override
                    public String apply(Object key, String value, String aggValue) {
                        System.out.println("spring集成以后的流式处理------------");
                        if(StringUtils.isEmpty(value)){
                            return aggValue;
                        }
                        List<String> strings = Arrays.asList(value.toLowerCase().split(","));
                        int count = 0;
                        for (String string : strings) {
                            if(string.equals("hello")){
                                count++;
                            }
                        }
                        return String.format("hello:%d",count);
                    }
                },Materialized.as("kafka-stream-sample-count-1")).toStream()
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), value);
                });
    }
}
